Kendraで検索対象を絞る2つの方法【アクセス制御・属性フィルタ】

山本紘暉

2023.12.04

はじめに

新規事業部生成AIチーム山本です。

Amazon KendraはAWSが提供しているエンタープライズ検索サービスです。数多くのドキュメントサービスと接続できるコネクタがあり、取得したデータを一括で検索できるというメリットがあります。最近では、生成AIに独自の情報（社内ドキュメントなど）に関して質問を行うための、Retrieval Augmented Generation（RAG）の検索部分として使われるケースが多いです。

Kendraのような検索サービスを使っていると、範囲を絞った特定のドキュメント内で検索を実行したいときがあります。例えば以下のようなケースです。

ある業務に関連する情報だけがほしい（その業務のドキュメントのみを調べたい）
権限的にある人には見せたくないドキュメントがある（そのドキュメントは検索対象から外したい）

この記事では、上記のような検索を実現するための、Kendraの設定方法について記載します。

※ Kendraの基本的な使い方については記載していませんので、他の記事をご参照ください

追記（2024/05/07）

_source_uriの形式が変わっているようです。ご注意ください。 - 本記事内（変更前）：s3://bucketName/contents/.... - 変更後：https://<バケット名>.s3.ap-northeast-1.amazonaws.com/...

前提の説明

方針

こうしたとき、設定の方針として以下の3つが考えられます。

方針１：検索対象ごとに分けてインデックスを作成する（インデックスを複数作成する）
方針２：呼び出した側で、条件にあったドキュメントのみに絞り込む（アプリ側で検索後に処理をする）
方針３：検索時に対象を設定する

方針１は、2つの問題があります。まずコストがかかる点が問題です。特に、Kendraはインデックスごとに課金されるため、分ければ分けるほど料金が増えてしまいます（容量で課金されるエンタープライズ検索サービスの場合なら、この点は解消されます）。また、検索精度にも問題が出る可能性があります。複数の対象から検索をするときに、インデックスごとに検索結果（=ランキング）が分かれてしまうため、あるインデックスの中に他よりも関連度が高いドキュメントがあったときにそれを見逃す可能性があります（また、複数の検索結果を統合する処理を実装するのも手間です）。

方針２でも目的は達成できますが、インデックス内に関係ないドキュメントがたくさん入っていると、1回の検索でドキュメントがヒットせず、何度も実行する必要があると予測されます。そうすると検索に時間がかかる点や、クエリレートを消費してしまうため、効率が悪くなる点がデメリットとして挙げられます。

方針３のように、検索時に対象を設定することで、1つのインデックスでコストを抑え、かつ、1回の実行で効率よく検索したいです。この利用方法として、Kendraには以下2つの方法があります。

属性フィルタ（AttributeFilter）
アクセス制御（AccessControl）

この記事では、それぞれの方法の概要と使い方について説明します。使い方としては、S3バケット内に3つのフォルダを作成し、いずれか1つのフォルダ内のみを対象として検索できるように設定するケースを例に説明します。（※ フォルダ：正確にはキーのプレフィクスのことです）

想定するユースケース例

今回は、説明ための簡単な例として、以下のような3つのシステムを作成することを想定します。

総務用検索システム
法務用検索システム
労務用検索システム

それぞれ用のドキュメントが決まっていて、フォルダごとに分けられている想定です。また、Kendraのインデックスは1つのみ使用し、対応するドキュメントのみを対象とした検索を行います。

以下は、データソースとしてS3バケットを使う場合について記載します。（未調査ですが、他のデータソースでも基本的な考え方・操作方法は同様だと思われます）

※ 今回は、S3バケットにドキュメントをアップロードする際に、プレフィクスにcontents/をつける場合について記載します。contents/をつけなくても、アクセス制御・属性フィルタは使用可能ですが、後述のようにS3バケットにアクセス制御の設定ファイルもアップロードする必要があり、それをこのバケットを1つで済ませたいため、ドキュメントのフォルダを分けました。（S3バケットをもう1つ用意して、そこにアクセス制御の設定ファイルをアップロードし、コンテンツ用のバケットにはトップのフォルダからドキュメントを置く、という構成も可能です）

概要

Kendraで読み込んだドキュメントには、メタデータとして属性（Attribute）が自動で付与されていて、この条件を指定して検索対象を絞ることができます。属性ごとに条件付けをして、組み合わせた論理条件をもとに検索できる、という仕組みになっています。ただし、属性フィルタだけだと今回のユースケースの要件を達成できないため、Custom Document Enrichment（以下、CDE）を使ってドキュメントをタグ付けする必要があります。

具体的には

属性の種類には、以下の4種類があります。（使用できる属性は、インデックスのページの「Facet definition」から確認できます）。

文字列（String）
文字列リスト（StringList）
数値（Long）
日付（Date）

検索時にそれぞれの属性に対して条件を指定できます。用意されているのは、以下のものです。これらを各属性に対する条件として使用可能です。

https://docs.aws.amazon.com/kendra/latest/APIReference/API_AttributeFilter.html#API_AttributeFilter_Contents

EqualsTo（全ての属性種類で利用可能、ドキュメントの属性の値と等しい）
ContainsAll（StringListで利用可能、指定したstringのlist=複数の文字列のすべてが、ドキュメントの属性の値に含まれている）
ContainsAny（StringListで利用可能、指定したstringのlist=複数の文字列のいずれかが、ドキュメントの属性の値に含まれている）
GreaterThan（DateとLongで利用可能、ドキュメントの属性の値より大きい）
GreaterThanOrEquals（DateとLongで利用可能、ドキュメントの属性の値以上）
LessThan（DateとLongで利用可能、ドキュメントの属性の値より小さい）
LessThanOrEquals（DateとLongで利用可能、ドキュメントの属性の値以下）

これらをNOTを取ることも可能です

NotFilter（条件の逆を取る）

複数をANDやORで階層的に組み合わせることも可能です。

AndAllFilters（すべての条件が成立したら、検索対象にする）
OrAllFilters（いずれかの条件が成立したら、検索対象にする）

例えば、「作成者に”user_a”が含まれる」かつ「作成日時が2023年12月4日以降」みたいな条件を設定できます。

今回のユースケースの場合

要件を実現するためには「ドキュメントのパスが”s3://bucketName/contents/soumu”で始まるもの」のような条件づけをしたいです。ドキュメントのパスは「_source_uri」に格納されているので、これを検索条件に設定すれば良さそうです。

ただし、上記のように「_source_uri」は属性種類がStringであるため、EqualsToしか条件として利用できず、目的の検索ができません。そのため、CDEを使用する必要があります。

CDEでは検索条件よりも複雑な設定が可能です。例えば、Stringの属性が「〇〇で始まる」とき、ある属性に文字列を追加する、といった操作が可能です。これを利用して以下のような流れで、ドキュメントにタグ付けすることで、対象を絞った検索ができるようになります。

実行例

インデックスに属性を追加する

インデックスに新しい属性を追加します。先程の「Facet Definition」のページの「Index fields」の「Add field」ボタンを押します。今回はField nameにgroup_tagという名前を指定し、Data typeをStringを指定します。Field nameは別のものでも構いません。「Add」ボタンを押して追加します。

※ 今回はStringを指定しましたが、StringListも候補としてありうると考えられます。例えば、共通のドキュメントがあり、それはどちらの検索条件のときでも対象にしたい場合です。StringListにタグを複数登録するようにCDEの条件を設定し、検索時に1つタグを指定する方法です。

データソースにCDEを設定する

Kendraのページでインデックスを選択し、左ペインの「Enrichment」の「Document enrichments」を選択します。「Add document enrichment」ボタンを押します。

「Document enrichment source」で対象のS3データソースを選択します。「Configure basic operations」で、以下のように設定します。以下はsoumuの場合で、houmu・roumuの分を繰り返します。

Document field name：「_source_uri」を選択
Condition operator：「BeginsWith」を選択
Condition value：「{"StringValue": "s3://bucketName/contents/soumu"}」と入力
Index field name：「group_tag」を選択
Target value：「{"StringValue": "soumu"}」と入力
Target action：「Update」を選択

※ source_uriの形式が変更されているようです。s3://bucketName/contents/soumuは、https://bucketName.s3.ap-northeast-1.amazonaws.com/soumuになるかと思います（2024/05/07追記）

「Next」を押し、次のページでも「Next」を押し、最後の確認画面で「Add document enrichment」を押します

これによって、データソースでSyncを実行したときに、自動でこの設定が実行されます。つまり、ドキュメントのS3のパスに応じたテキストがgroup_tagに追加された状態で、インデックスに格納されます。

対象のデータソースでSycnを実行します。すでにSyncを実行している場合も、この設定を反映させるためにSyncを実行する必要があります。

検索時に属性フィルタを設定する

queryメソッドのAttributeFilterに、group_tagに関する条件を追加します。コードとしては以下のとおりです。

import boto3

kendra = boto3.client("kendra")

tag_name = "soumu"  # アプリごとに変えて設定する想定（環境変数などで与える/URLパスで指定する）

kendra.query(
    QueryText="日比谷オフィスの解錠方法",
    IndexId="xxxxxxxx-xxxx-xxxx-xxxxxxxxxxxxxxxxx",
    AttributeFilter={
        "AndAllFilters": [
            {
                "EqualsTo": {
                    "Key": "_language_code",
                    "Value": {
                        "StringValue": "ja",  # Sync時の言語に応じて変更してください
                    },
                },
            },
            {
                "EqualsTo": {
                    "Key": "group_tag",
                    "Value": {
                        "StringValue": tag_name,
                    },
                },
            },
        ]
    },
    DocumentRelevanceOverrideConfigurations=[
        {
            "Name": "test",
            "Relevance": {
                "RankOrder": "DESCENDING",
            },
        }
    ],
    PageSize=10,
)

アクセス制御（AccessControl）

概要

以下のように設定し検索します。

設定：ユーザ名やグループごとにアクセスできるドキュメントを決める
アクセスのルールを書いたAccess Control List（以下、ACL）を作成し、データソースのパラメータとして設定します（Sync時に適用されます）。ACLは以下の3つについて設定できます。
- プレフィクスごとに（フォルダ・ファイルごと）
- ユーザ・グループごとに
- アクセスを許可するか・禁止するか
検索：検索パラメータにユーザの情報を設定する
ユーザの情報は以下の2つの渡し方があります
- ユーザ名・グループ名を直接パラメータで指定する（アプリ側で認証情報を取得して、検索のパラメータとして渡す）
- ユーザトークンをパラメータで指定する（認証情報の検証、ユーザ名・グループ名の取得をKendraに任せる）

実行例

ACL

ACLはjsonファイルとして作成します。前述のユースケース例を実装する場合、以下のように記載します。各ドキュメントに対するアクセスを、グループとして許可するという設定内容です。これにより、各グループとして検索を行ったとき、対応するフォルダ内部のみのドキュメントを検索するようになります。

※ 「Type」を「USER」として設定することも可能ですが、このケースだと「総務グループの人はアクセス可能」という考え方が適切と思われるので、「GROUP」としました。「USER」でも対象を絞った検索は可能です。

※ 各keyPrefixのbucketNameは、お使いの環境のS3バケット名に合わせて変更してください。

[
    {
        "keyPrefix": "s3://bucketName/soumu/",
        "aclEntries": [
            {
                "Name": "soumu_group",
                "Type": "GROUP",
                "Access": "ALLOW"
            }
        ]
    },
    {
        "keyPrefix": "s3://bucketName/houmu/",
        "aclEntries": [
            {
                "Name": "houmu_group",
                "Type": "GROUP",
                "Access": "ALLOW"
            }
        ]
    },
    {
        "keyPrefix": "s3://bucketName/roumu/",
        "aclEntries": [
            {
                "Name": "roumu_group",
                "Type": "GROUP",
                "Access": "ALLOW"
            }
        ]
    }

]

これをS3のバケットにアップロードし、データソースを作成するときの「Access control list configuration file location」としてパスを設定します。以下は、バケットのacl/acl.jsonとしてアップロードした場合です（他のパスでも可能です）。

※ 図中のバケット名を隠すため、上からテキストを貼り付けています

検索パラメータ

前述のとおり、以下の2通りが使用可能です。

ユーザ名・グループ名を直接パラメータで指定する

検索を呼び出す側（アプリ）で、予めアクセスしたユーザのユーザ名やグループ名がわかっていることが前提です。このユーザ名やグループ名をqueryメソッドのUserContextパラメータに、Groupsとして指定します。サンプルコードとしては、以下のとおりです。

import boto3

kendra = boto3.client("kendra")

group_name = "soumu_group"  # ここが予めわかっている想定

kendra.query(
    QueryText="日比谷オフィスの解錠方法",
    IndexId="xxxxxxxx-xxxx-xxxx-xxxxxxxxxxxxxxxxx",
    AttributeFilter={
        "EqualsTo": {
            "Key": "_language_code",
            "Value": {
                "StringValue": "ja",  # Sync時の言語に応じて変更してください
            },
        }
    },
    UserContext={"Groups": [group_name]},
    DocumentRelevanceOverrideConfigurations=[
        {
            "Name": "test",
            "Relevance": {
                "RankOrder": "DESCENDING",
            },
        }
    ],
    PageSize=10,
)

ユーザトークンをパラメータで指定する
この方法を使用するには、認証サーバを使用する必要があります。今回はCognitoを認証サーバとして使用するケースについて説明します。（未調査ですが、他の認証サーバ・IdPを使う場合も、基本的な考え方は同じだと思います）

設定：
- Cognitoでユーザプールを作成します。総務・法務・労務に対応するグループを作成します。今回のアプリで使う、アプリケーションクライアントを作成します
- Kendraのインデックスの設定として、「Access control settings」でyesを選択し、「Token configuration」では、「Token type」で「OpenID」を選択し、「Signing key URL」で「https://cognito-idp.(リージョン名).amazonaws.com/(作成したCognitoのユーザプールID)/.well-known/jwks.json」を指定します
- 「Advanced configuration」を開き、「Username」に「cognito:username」を、「Groups」に「cognito:groups」を入力します（注：2023/12/05 11:00追記しました）
準備：
- ユーザがCognitoに登録（サインアップ）します（もしくは、管理者がCogintoのユーザとして登録します）
- 管理者がグループを作成し、ユーザをグループに加えます
使用：
- ユーザ（Kendraを呼び出すアプリ）はCognitoにサインインして、トークン（JWT）をCognitoから受け取ります（認証）
- これをqueryメソッドのUserContextパラメータに、Tokenとして指定します（検索）
- KendraはこれをCognitoのキーを使って、正しく発行されたものか確認します（検証）（この部分は自動で実行されます）
コードとしては以下のとおりです。（実際のアプリ・サービスでは、フロントエンドとバックエンドとして分かれる部分かと思います）
- トークン取得部分
```
import boto3

cognito = boto3.client('cognito-idp')

USER_POOL_ID = "ap-northeast-1_XXXXXXXXX"
CLIENT_ID = "xxxxxxxxxxxxxxxxxxxxxxxxxx"

username = ""  # どこかから取得する
password = ""  # どこかから取得する

auth_info = cognito.admin_initiate_auth(
    UserPoolId = USER_POOL_ID,
    ClientId = CLIENT_ID,
    AuthFlow = "ADMIN_NO_SRP_AUTH",
    AuthParameters = {
        "USERNAME": username,
        "PASSWORD": password,
    }
)

token = aws_result["AuthenticationResult"]["IdToken"]
```
- 実行部分
```
import boto3

kendra = boto3.client("kendra")

token = "eyJ...XXX"  # 上のコードのtokenを受け取る想定

kendra.query(
    QueryText="日比谷オフィスの解錠方法",
    IndexId="xxxxxxxx-xxxx-xxxx-xxxxxxxxxxxxxxxxx",
    AttributeFilter={
        "EqualsTo": {
            "Key": "_language_code",
            "Value": {
                "StringValue": "ja",  # Sync時の言語に応じて変更してください
            },
        }
    },
    UserContext={"Token": token},
    DocumentRelevanceOverrideConfigurations=[
        {
            "Name": "test",
            "Relevance": {
                "RankOrder": "DESCENDING",
            },
        }
    ],
    PageSize=10,
)
```

注意点

アクセス制御は一部のコネクタでは対応していません。つまり、検索したときに、そのコネクタでインポートしたドキュメントが検索対象となってしまうため（本来対象としたいドキュメント以外が検索対象となってしまうため）、検索の邪魔になってしまうケースがあります。

どのコネクタが対応しているかは、以下のページから調べることができます。各コネクタのページの「Supported features」に「User context filtering」があれば、アクセス制御は使用可能です。

Data source connectors - Amazon Kendra

（S3を含む）ほとんどのコネクタでアクセス制御は使用可能ですが、対応していないコネクタの一つにweb crawlerがあり、このアクセス制御が使えない点に注意した方が良いかと思います。

（元々、接続先のサービスのトークンを使って、そのサービスのアクセス制御設定を使う、という利用が想定されていると思われます。Webの情報にアクセス制御という概念は当てはまらないので、Web Crawlerは対応していない、ということかなと思います）

どちらの機能を使うべきか

簡単な振り分けをしたい場合

上記のような振り分けの条件が簡単な場合、どちらの方法を使っても実現可能です。上記のように各部門の専用のボットをつくる場合など、多くのケースがこれに該当すると思われます（もう少し大まかに言うと、要件が、同じKendraのインデックスを共有してコストを抑えることのみ、である場合です）。

Web Crawlerのデータソースを使う場合など、アクセス制御が対応していない場合は、属性フィルタを利用する必要があります。

概念的には、アクセス制御はユーザに対して行うもの、属性フィルタはコンテンツに対して行うものです。なので、どちらも利用できる場合は、これを考慮にいれて採用した方が、混同しなくて良いかと思います。

複雑な条件づけをしたい場合

もし大規模で複雑な使い方をするようになってくると、2つの方法を組み合わせて利用すると良さそうです。例えば、以下のような要件をすべて満たす必要があるときです。

1つのKendraインデックスでコストを抑えたい（複数にインデックスを作りたくない）
ユーザの所属や職務レベルによって見られるドキュメントを制限したい（ユーザに合わせた適切なアクセス制御をしたい）
検索時にはさまざまな条件で検索を試行錯誤したい（必要に応じて検索条件を変えたい）
さまざまなデータソースから取得したドキュメントを横断的に検索できるようにしたい（Kendraのコネクタを使用したい）

このときは、2点目をアクセス制御に任せ、3点目を属性フィルタに任せる、という使い分けが良さそうです。

※ 補足：様々な条件で試行錯誤するという点について。今後、Agentのような役割が広く使われるようになり、自動でさまざまな条件で検索を行ってくれるようになることが予想されます。こうしたときに、指示を出すユーザが閲覧できるドキュメントの範囲をアクセス制御で決めておいて、その中で属性フィルタを使って自動で何度も検索する、みたいな使い方になるのかなと思われます。

まとめ

Kendraの検索で、対象とするドキュメントを絞る方法として、属性フィルタ・アクセス制御の2つの機能を紹介し、それぞれの設定方法を説明しました。

書いていたら意外と長くなってしまいました。読んでくださってありがとうございます。

きっとだれかの役にたつはず

Appendix・補足・付録

CDEのさらなる機能

今回はCDEの「Basic operations」という機能を使用しました。この機能では、BeginsWithのような既に用意されている条件を指定可能です。さらに複雑な条件で判定してタグ付けを行いたい場合、CDEからLambda関数に渡して処理することも可能です。Lambda関数なので、ライブラリを使ったり、複雑なロジックを使って判定させることができます（また、ドキュメントの読み込み方をロジックを自分で実装したい場合にも使用できます）。今回のような要件では不要なため使用しませんでしたが、このあたりも試してみたいです。

Signing key URLの中身と検証の仕組み

上記のhttps://…/jwks.jsonにアクセスすると、以下のようなJSONファイルを取得できます（一部省略しています）。これは公開鍵になっていて、これによりユーザトークン（JWT）の署名部分を使ってユーザ情報を検証できます。このアドレスをKendraのSigning key URLに指定することで、Kendraはトークンがそのユーザプールから発行されたこと（=そのユーザプールで設定された内容であること）が確認でき、ユーザIDやグループの設定を正しく利用できる（不正な利用を防げる）、という仕組みです。

{
  "keys": [
    {
      "alg": "RS256",
      "e": "AQAB",
      "kid": "LeF0...TDQ=",
      "kty": "RSA",
      "n": "seRfHr...0vu-88tF4KQ",
      "use": "sig"
    },
    {
      "alg": "RS256",
      "e": "AQAB",
      "kid": "JOv...xr4FHlc=",
      "kty": "RSA",
      "n": "wGaMxa...HDKTKM03dA7ALC6Q",
      "use": "sig"
    }
  ]
}

UserContextの設定パラメータ

UserContextの引数の型定義を見ると、以下のようになっていました。これをみると以下の点が気になります。

1点目はグループを複数指定可能なことです。おそらくですが、複数のグループに所属するユーザに対しては、それらのグループで許可・禁止されている複数の条件が組み合わさって検索される、という機能のように思えます。
2点目はDataSourceGroupsとは何か、です。検索してもあまり情報が出てこなかったので、どういう機能なのか・どういう使い方をするのか知りたいところです。
3点目はTokenやUserIdやGroupsやDataSourceGroupsを同時に指定できるのか、という点です。Tokenは前述のように、ユーザのIDやグループの情報をもたせることができるため、UserIDやGroupsと同時に使用すると、設定がバッティングしそうです。また、UserIdとGroupsは同時に設定可能に思われますが、ACLに両方の条件に当てはまるルールが書いて有ったときは、どちらの設定が優先されるのか、が気になります。（おそらく、AWSのIAM考え方と同じだとすると、ALLOWよりもDENYが優先されるのだと思われます）

（このあたりは、今後調べて記事にできればと思っています）

UserContextTypeDef = TypedDict(
    "UserContextTypeDef",
    {
        "Token": NotRequired[str],
        "UserId": NotRequired[str],
        "Groups": NotRequired[Sequence[str]],
        "DataSourceGroups": NotRequired[Sequence[DataSourceGroupTypeDef]],
    },
)